/*
 * Copyright 2009 Armando Blancas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package taskgraph.tasks.data;

import taskgraph.Config;
import taskgraph.channels.Channel;
import taskgraph.ports.InputPort;
import taskgraph.ports.OutputPort;
import taskgraph.tasks.CompositeTask;

/**
 * Find distinct objects across the whole input. This is in contrast
 * to task {@code Unique}, which works locally to find distinct or unique
 * objects by comparing one to the next.
 *  
 * @author Armando Blancas
 * @see taskgraph.tasks.data.Sort
 * @see taskgraph.tasks.data.Unique
 */
public class GlobalUnique<E> extends CompositeTask {

	private static final int DEFAULT_CAPACITY = Config.get().channelCapacity();
	
	private InputPort<E> input;
	private OutputPort<E> output;
	private int bufferSize = DEFAULT_CAPACITY;

	/**
	 * This constructor allows creating instances as beans.
	 */
    public GlobalUnique() {
    	// do nothing
    }

	/**
     * Creates a fully configured instance.
     * 
     * @param input The input port with elements of type E.
     * @param output The output port with elements of type E.
     */
    public GlobalUnique(final InputPort<E> input, 
    		            final OutputPort<E> output) {
    	setInput(input);
    	setOutput(output);
    }

	/**
	 * Gets the buffer size for the internal channels.
	 * 
	 * @return the bufferSize
	 */
	public int getBufferSize() {
		return bufferSize;
	}

    /**
     * Gets the intput port with elements of type E.
     * 
	 * @return The input port.
	 */
	public InputPort<E> getInput() {
		return input;
	}
	
    /**
     * Gets the output port with elements of type E.
     * 
	 * @return The output port.
	 */
	public OutputPort<E> getOutput() {
		return output;
	}

	/**
	 * Sets the buffer size for the internal channels.
	 *  
	 * <p>The argument should not be less than the default. If it is, the
	 * default size is used instead. Once set, channels will be able to
	 * hold a number of object references equal to {@code bufferSize}.
	 * 
	 * @param bufferSize The bufferSize to set.
	 */
	public void setBufferSize(final int bufferSize) {
		this.bufferSize = 
			(bufferSize < DEFAULT_CAPACITY) ? DEFAULT_CAPACITY : bufferSize;
	}

    /**
	 * Sets the input port.
	 * 
	 * @param input The input port.
	 */
    public void setInput(final InputPort<E> input) {
    	if (input == null)
    		throw new IllegalArgumentException("input == null");
 		this.input = input;
	}
    
	/**
	 * Sets the output port.
	 * 
	 * @param output The output port.
	 */
    public void setOutput(final OutputPort<E> output) {
    	if (output == null)
    		throw new IllegalArgumentException("output == null");
 		this.output = output;
	}

	/**
     * Creates a graph that finds distinct objects across the whole input
     * by first sorting all of the the incoming data.
     * <pre>
     * --input--> [Sort] --ch1--> [Unique] --output-->
     * </pre>
	 */
	@Override
    protected void setup() {
		Channel<E>  ch = new Channel<E>(bufferSize);
		
        add(new Sort<E>(input, ch.getOutputPort()));
        add(new Unique<E>(ch.getInputPort(), output));
    }

}
